package core_sql.day03

import java.net.URL

import util.FileUtil
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by root on 2019/3/10.
  *
  */
object SubjectFavTeacher2 {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("subjectTeacher").setMaster("local[*]")

    val sc: SparkContext = new SparkContext(conf)

    val lines: RDD[String] = sc.textFile(FileUtil.TEACHER_LOG)

    val subjectTeacherAndOne: RDD[((String, String), Int)] = lines.map(line => {
      val index = line.lastIndexOf("/") + 1
      val teacher = line.substring(index)
      //切分学科
      val host = new URL(line).getHost
      val subject: String = host.substring(0, host.indexOf("."))
      ((subject, teacher), 1)
    })

    //聚合
    val reduced: RDD[((String, String), Int)] = subjectTeacherAndOne.reduceByKey(_+_)

    //在学科累不进行排序
    val grouped: RDD[(String, Iterable[((String, String), Int)])] = reduced.groupBy(_._1._1)

    //分组之后进行内部排序（组内排序）
    val result: RDD[(String, List[(String, Int)])] = grouped.mapValues(_.toList.sortBy(_._2).reverse.take(2).map(x=>(x._1._2,x._2)))

    result.foreach(println(_))

    sc.stop()

  }

}
